package mr.dividends;

import java.io.IOException;

import mr.dividends.model.DividendsKeyWritable;
import mr.dividends.model.DividendsValueWritable;

import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DividendsCombiner extends Reducer<DividendsKeyWritable, DividendsValueWritable, DividendsKeyWritable, DividendsValueWritable> {

	private static Logger LOG = LoggerFactory.getLogger(DividendsCombiner.class); 
	  
	@Override
    protected void reduce(DividendsKeyWritable key, Iterable<DividendsValueWritable> values, Context context) throws IOException,
            InterruptedException {
		DividendsValueWritable value =new DividendsValueWritable(); 
		for(DividendsValueWritable dividends : values){
			LOG.info("Combiner in action! " + key.toString() + Float.toString(dividends.getAnnualDividendsTotal()));
			value.setAnnualDividendsTotal(value.getAnnualDividendsTotal() + dividends.getAnnualDividendsTotal());
			value.setAnnualCount(value.getAnnualCount() + dividends.getAnnualCount());
		}
		LOG.info("Combiner in action end"); 
		context.write(key, value);
	}
}
